package com.clown.LoginFailDetect

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

object LoginFailAdvance {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val resource = getClass.getResource("/LoginLog.csv")
    val inputStream = env.readTextFile(resource.getPath)
    // 转换成样例类类型，并提取时间戳和watermark
    val loginEventStream = inputStream
      .map(data => {
        val arr = data.split(",")
        LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(2)) {
        override def extractTimestamp(element: LoginEvent): Long = element.timestamp * 1000L
      })

    // 进行判断和检测，如果2秒之内连续登陆失败，输出报警信息
    val loginFailWarningStream = loginEventStream
      .keyBy(_.userId)
      .process(new LoginFailWarningAdvanceResult(2, 2))

    loginFailWarningStream.print()
    env.execute("login fail warning job")
  }

}

class LoginFailWarningAdvanceResult(failTimes: Int, second: Int) extends KeyedProcessFunction[Long, LoginEvent, LoginFailWarning] {
  lazy val loginFailState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("loginFailState", classOf[LoginEvent]))
  lazy val timerTimestampState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timerTimestampState", classOf[Long]))

  override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#Context, out: Collector[LoginFailWarning]): Unit = {
    // 判断当前登录事件是成功还是失败
    if (value.eventType.equals("fail")) {
      // 1. 如果是失败，进一步判断
      val iterator = loginFailState.get().iterator()
      // 判断之前是否有登录事件
      if (iterator.hasNext) {
        // 1.1 如果有，那么判断两次失败的时间差
        val fisrtFailEvent = iterator.next()
        if (value.timestamp < fisrtFailEvent.timestamp + second) {
          // 如果在n秒之内，输出报警
          out.collect(LoginFailWarning(value.userId, fisrtFailEvent.timestamp, value.timestamp, "login fail %s times in %ss".format(failTimes, second)))
        }
        // 不管报警不报警，当前都已处理完毕，将状态更新为最近依次登录失败的事件
        loginFailState.clear()
        loginFailState.add(value)
      } else {
        // 1.2 如果没有，直接把当前事件添加到listState中
        loginFailState.add(value)
      }
    } else {
      // 如果是成功，那么直接清空状态
      loginFailState.clear()
    }
  }
}